package com.kamistoat.rocketmqlearning.transactionMsg;

import com.kamistoat.rocketmqlearning.common.R;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.nio.charset.StandardCharsets;

@RequestMapping("/transactionMsg")
@Controller
public class TransactionMsgController {
    @Autowired
    @Qualifier("transactionMsgProducer")
    TransactionMQProducer msgProducer;

    @RequestMapping("/sync")
    @ResponseBody
    public R syncProducer() {
        try {
            String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 5; i++) {
                byte[] body = ("Hi " + i).getBytes(StandardCharsets.UTF_8);
                Message mQMsg = new Message("transTopic", tags[i], body);
                /**
                 * 第二个参数是执行Listener中本地事务时传入的参数
                 */
                TransactionSendResult transactionSendResult = msgProducer.sendMessageInTransaction(mQMsg, null);
                System.out.println(transactionSendResult);
            }
            return R.ok();
        } catch (MQClientException e) {
            e.printStackTrace();
            return R.error(e.getMessage());
        }
    }
}
